package org.sg20.parsename;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class ParseNameACAutomaton {

    public static class ParseNameACAutomatonMapper extends Mapper<LongWritable, Text, ParseNameOutputKey, Text> {
        ParseNameOutputKey outkey;
        Text outvalue;
        ACAutomaton nametree = new ACAutomaton();

        @Override
        protected void setup(Mapper<LongWritable, Text, ParseNameOutputKey, Text>.Context context) throws IOException {
            Path nameList = new Path(context.getCacheFiles()[0]);
            FileSystem fs = FileSystem.get(context.getConfiguration());
            InputStreamReader isr = new InputStreamReader(fs.open(nameList));

            try {
                BufferedReader br = new BufferedReader(isr);
                // read until end of file
                String name;
                while ((name = br.readLine()) != null) {
                    // add the name to the trie tree
                    nametree.insertWord(name);
                }
                br.close();
                nametree.buildGraph();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, ParseNameOutputKey, Text>.Context context) throws IOException, InterruptedException {
            String words = value.toString();
            String filename = ((FileSplit) context.getInputSplit()).getPath().getName();
            String res = nametree.parseString(words);

            outkey = new ParseNameOutputKey(filename, key.get());
            outvalue = new Text(res);
            if(!res.equals(""))
                context.write(outkey, outvalue);
        }
    }

    public static class ParseNameACAutomatonReducer extends Reducer<ParseNameOutputKey, Text, ParseNameOutputKey, Text> {
        // There is nothing to do for reducer
        @Override
        protected void reduce(ParseNameOutputKey key, Iterable<Text> values, Reducer<ParseNameOutputKey, Text, ParseNameOutputKey, Text>.Context context) throws IOException, InterruptedException {
            super.reduce(key, values, context);
        }
    }

}
